package com.alibaba.dubbo.demo.filter;

import java.lang.reflect.Field;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.Invocation;
import org.apache.dubbo.rpc.Invoker;
import org.apache.dubbo.rpc.ListenableFilter;
import org.apache.dubbo.rpc.Result;
import org.apache.dubbo.rpc.RpcContext;
import org.apache.dubbo.rpc.RpcException;
import org.springframework.util.ReflectionUtils;

import com.alibaba.dubbo.rpc.support.RpcUtils;

import reactor.core.publisher.MonoSink;

@Activate(group = CommonConstants.CONSUMER, order = 10)
public class ReactiveConsumerFilter extends ListenableFilter {

	private static final Map<Long, Object> SINKS = new ConcurrentHashMap<>();

	public ReactiveConsumerFilter() {
        super.listener = new ReactiveConsumerListener();
    }

	@Override
	public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
		Long id = RpcUtils.getInvocationId(invocation);
		if(id == null) {
			RpcUtils.attachInvocationIdIfAsync(invoker.getUrl(), invocation);
			id = RpcUtils.getInvocationId(invocation);
		}
		// 把sink全局缓存
		SINKS.put(id, RpcContext.getContext().get("SINK"));
		return invoker.invoke(invocation);
	}

	class ReactiveConsumerListener implements Listener {

		@Override
		public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
			dealSink(invocation);
		}

		/**处理sink
		 * @param invocation
		 */
		private void dealSink(Invocation invocation) {
			Long id = RpcUtils.getInvocationId(invocation);
			if(id != null) {
				Object sinkObj = SINKS.remove(id);
				if(sinkObj instanceof MonoSink) {
					MonoSink<?> sink = (MonoSink<?>)sinkObj;

					// 拷贝sink对象和class对象，同时要求回调方法必须用带参数的那种
					copyArg(invocation, sink);
					copyType(invocation, sink);
				}
			}
		}

		/**sink对象加入到arguments
		 * @param invocation
		 * @param sink
		 */
		private void copyArg(Invocation invocation, MonoSink<?> sink) {
			Field argumentsField = ReflectionUtils.findField(invocation.getClass(), "arguments");
			ReflectionUtils.makeAccessible(argumentsField);
			Object[] arguments = invocation.getArguments();
			Object[] newArguments = new Object[arguments.length + 1];
			newArguments[0] = sink;
			System.arraycopy(arguments, 0, newArguments, 1, arguments.length);
			ReflectionUtils.setField(argumentsField, invocation, newArguments);
		}

		/**sink的class类型加入parameterTypes
		 * @param invocation
		 * @param sink
		 */
		private void copyType(Invocation invocation, MonoSink<?> sink) {
			Field parameterTypesField = ReflectionUtils.findField(invocation.getClass(), "parameterTypes");
			ReflectionUtils.makeAccessible(parameterTypesField);
			Class<?>[] parameterTypes = invocation.getParameterTypes();
			Class<?>[] newParameterTypes = new Class<?>[parameterTypes.length + 1];
			newParameterTypes[0] = sink.getClass();
			System.arraycopy(parameterTypes, 0, newParameterTypes, 1, parameterTypes.length);
			ReflectionUtils.setField(parameterTypesField, invocation, newParameterTypes);
		}

		@Override
		public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
			dealSink(invocation);
		}

	}

}
